package com.tianji.remark.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tianji.api.dto.remark.ChangedTimesDTO;
import com.tianji.common.autoconfigure.mq.RabbitMqHelper;
import com.tianji.common.utils.CollUtils;
import com.tianji.common.utils.StringUtils;
import com.tianji.common.utils.UserContext;
import com.tianji.remark.constants.RedisConstants;
import com.tianji.remark.domain.dto.LikeRecordFormDTO;
import com.tianji.remark.domain.dto.LikedRecordsDTO;
import com.tianji.remark.domain.po.LikedRecord;
import com.tianji.remark.mapper.LikedRecordMapper;
import com.tianji.remark.service.ILikedRecordService;
import com.tianji.remark.task.LikedTimesAndRecordsCheckTask;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static com.tianji.common.constants.MqConstants.Exchange.LIKE_RECORD_EXCHANGE;
import static com.tianji.common.constants.MqConstants.Key.LIKED_RECORDS_KEY_TEMPLATE;
import static com.tianji.common.constants.MqConstants.Key.LIKED_TIMES_KEY_TEMPLATE;

/**
 * <p>
 * 点赞记录表 服务实现类
 * </p>
 *
 * @author huanmo
 * @since 2025-09-22
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class LikedRecordServiceRedisImpl extends ServiceImpl<LikedRecordMapper, LikedRecord> implements ILikedRecordService {

    private final RabbitMqHelper mqHelper;
    private final StringRedisTemplate redisTemplate;
    private final LikedTimesAndRecordsCheckTask likedCheckTask;

    /**
     * 使用 Redis 的 Set 结构存储每个业务对象 (bizId) 的点赞用户 ID 集合，键格式为LIKES_BIZ_KEY_PREFIX + bizId
     * 使用 Redis 的 SortedSet 结构存储每个业务类型 (bizType) 的点赞数量统计，键格式为LIKES_TIMES_KEY_PREFIX + bizType
     * 每一次点赞，会把点赞记录记在Redis的Set中，取消点赞就会把这条记录从Set删除。
     * 然后把Set里面的所有记录整合，计算出总的需要改变的点赞次数，记录在SortedSet中。
     * 最后把SortedSet的数据传给数据库，数据库从而变化点赞次数
     */
    @Override
    public void addLikeRecord(LikeRecordFormDTO recordDTO) {
        // 1.基于前端的参数，判断是执行点赞还是取消点赞
        boolean success = recordDTO.getLiked() ? like(recordDTO) : unlike(recordDTO);
        // 2.判断是否执行成功，如果失败，则直接结束
        if (!success) {
            return;
        }
        // 3.如果执行成功，统计点赞总数
        // 将tj_remark.liked_record表中的所有点赞记录缓存至redis之后，直接在redis中进行数量统计
        // KEY（bizId） VALUE(userId)
        // 点赞数
        Long likedTimes = redisTemplate.opsForSet()
                .size(RedisConstants.LIKES_BIZ_KEY_PREFIX + recordDTO.getBizId());
        // 取消点赞数
        Long unlikedTimes = redisTemplate.opsForSet()
                .size(RedisConstants.UNLIKES_BIZ_KEY_PREFIX + recordDTO.getBizId());
        // 处理可能的null值
        likedTimes = likedTimes == null ? 0L : likedTimes;
        unlikedTimes = unlikedTimes == null ? 0L : unlikedTimes;
        // 计算净点赞数（点赞数 - 点踩数）
        long netLikes = likedTimes - unlikedTimes;
        // 4.缓存点赞总数到redis
        // KEY（bizType） Member(bizId)  Score(likedTimes)
        redisTemplate.opsForZSet().add(
                RedisConstants.LIKES_TIMES_KEY_PREFIX + recordDTO.getBizType(),
                recordDTO.getBizId().toString(),
                netLikes
        );
        // 设置过期时间
        redisTemplate.expire(RedisConstants.LIKES_TIMES_KEY_PREFIX + recordDTO.getBizType(), Duration.ofMinutes(RedisConstants.REDIS_EXPIRE_TIME_MINUTES + new Random().nextInt(12)));
    }

    /**
     * 查询指定业务id的点赞状态
     */
    @Override
    public Set<Long> isBizLiked(List<Long> bizIds) {
        log.debug("开始查询点赞状态，业务ID数量: {}", CollUtils.size(bizIds));
        // 1.获取登录用户id
        Long userId = UserContext.getUser();
        if (userId == null) {
            userId = 2L;
            log.warn("用户上下文为空，使用测试用户ID: {}", userId);
        }
        if (CollUtils.isEmpty(bizIds)) {
            log.warn("业务ID列表为空");
            return Collections.emptySet();
        }
        // 2.先查询Redis
        Set<Long> redisResult = queryLikedFromRedis(bizIds, userId);
        // 3.如果Redis结果不完整，回退到数据库查询
        if (needFallbackToDb(bizIds, redisResult)) {
            log.info("Redis数据不完整，回退到数据库查询，用户ID: {}", userId);
            Set<Long> dbResult = queryLikedFromDb(bizIds, userId);
            // 4.将数据库结果写回Redis（缓存预热）
            warmUpRedisCache(dbResult, userId);
            return dbResult;
        }
        log.debug("点赞查询完成，用户 {} 点赞了 {} 个业务", userId, redisResult.size());
        return redisResult;
    }

    /**
     * 从Redis查询点赞状态
     * redis使用管道(Pipelining)提高查询速度
     */
    private Set<Long> queryLikedFromRedis(List<Long> bizIds, Long userId) {
        List<Object> objects = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection src = (StringRedisConnection) connection;
            for (Long bizId : bizIds) {
                if (bizId != null) {
                    String key = RedisConstants.RECORDS_BIZ_KEY_PREFIX + bizId;
                    src.sIsMember(key, userId.toString());
                }
            }
            return null;
        });
        return IntStream.range(0, objects.size())
                .filter(i -> (boolean) objects.get(i))
                .mapToObj(bizIds::get)
                .collect(Collectors.toSet());
    }

    /**
     * 从数据库查询点赞状态
     */
    private Set<Long> queryLikedFromDb(List<Long> bizIds, Long userId) {
        // 查询数据库中的点赞记录
        List<LikedRecord> dbRecords = lambdaQuery()
                .eq(LikedRecord::getUserId, userId)
                .in(LikedRecord::getBizId, bizIds)
                .list();
        return dbRecords.stream()
                .map(LikedRecord::getBizId)
                .collect(Collectors.toSet());
    }

    /**
     * 判断是否需要回退到数据库查询
     */
    private boolean needFallbackToDb(List<Long> bizIds, Set<Long> redisResult) {
        // 如果Redis查询结果数量远小于业务ID数量，可能数据不完整
        // 这里可以根据业务需求调整策略
        // 90%的阈值
        return redisResult.size() < bizIds.size() * 0.9;
    }

    /**
     * 缓存预热：将数据库结果写回Redis
     */
    private void warmUpRedisCache(Set<Long> likedBizIds, Long userId) {
        if (CollUtils.isEmpty(likedBizIds)) {
            return;
        }
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection src = (StringRedisConnection) connection;
            for (Long bizId : likedBizIds) {
                String key = RedisConstants.RECORDS_BIZ_KEY_PREFIX + bizId;
                src.sAdd(key, userId.toString());
                // 由于变更频繁，redis缓存的有效期应当减短
                src.expire(key, RedisConstants.REDIS_EXPIRE_TIME_MINUTES + new Random().nextInt(12));
            }
            return null;
        });
        log.debug("缓存预热完成，为用户 {} 预热了 {} 个点赞记录", userId, likedBizIds.size());
    }

    /**
     * 从"直接替换"改为"增量更新"
     */
    @Override
    public void readLikedTimesAndSendMessage(String bizType, int maxBizSize) {
        log.debug("开始处理业务类型 {} 的点赞次数同步，最大处理数量: {}", bizType, maxBizSize);
        // 1.读取并移除Redis中缓存的点赞总数
        String key = RedisConstants.LIKES_TIMES_KEY_PREFIX + bizType;
        Set<ZSetOperations.TypedTuple<String>> tuples = redisTemplate.opsForZSet().popMin(key, maxBizSize);
        if (CollUtils.isEmpty(tuples)) {
            return;
        }
        // 2.数据转换
        List<ChangedTimesDTO> list = new ArrayList<>(tuples.size());
        for (ZSetOperations.TypedTuple<String> tuple : tuples) {
            String bizId = tuple.getValue();
            Double changedTimes = tuple.getScore();
            if (bizId == null || changedTimes == null || changedTimes == 0) {
                continue;
            }
            list.add(ChangedTimesDTO.of(Long.valueOf(bizId), changedTimes.intValue()));
        }
        // 3.发送MQ消息
        try {
            mqHelper.send(
                    LIKE_RECORD_EXCHANGE,
                    StringUtils.format(LIKED_TIMES_KEY_TEMPLATE, bizType),
                    list);
            log.info("成功发送MQ消息 - 业务类型: {}, 数据量: {} 条", bizType, list.size());
        } catch (Exception e) {
            log.error("发送MQ消息失败 - 业务类型: {}, 数据量: {} 条", bizType, list.size(), e);
        }
    }

    @Override
    public void readLikedRecordsAndSendMessage(String bizType, int maxBizSize) {
        log.debug("开始处理点赞记录同步，最大处理数量: {}", maxBizSize);
        try {
            List<LikedRecordsDTO> likedRecords = new ArrayList<>();
            // 1. 获取点赞记录并包装
            String likePattern = RedisConstants.LIKES_BIZ_KEY_PREFIX + "*";
            Set<String> likeKeys = redisTemplate.keys(likePattern);
            log.debug("找到 {} 个点赞记录的Redis键", CollUtils.isEmpty(likeKeys) ? 0 : likeKeys.size());
            if (CollUtils.isNotEmpty(likeKeys)) {
                for (String key : likeKeys) {
                    if (likedRecords.size() >= maxBizSize) {
                        break;
                    }
                    // 从key中提取bizId：likes:set:biz:1585191151894343681 -> 1585191151894343681
                    String bizIdStr = key.substring(RedisConstants.LIKES_BIZ_KEY_PREFIX.length());
                    try {
                        long bizId = Long.parseLong(bizIdStr);
                        // 获取该bizId下的所有userId
                        List<String> userIds = redisTemplate.opsForSet().pop(key, maxBizSize - likedRecords.size());
                        if (userIds != null && !userIds.isEmpty()) {
                            for (String userIdStr : userIds) {
                                Long userId = Long.valueOf(userIdStr);
                                LikedRecordsDTO record = LikedRecordsDTO.builder()
                                        .userId(userId)
                                        .bizId(bizId)
                                        .liked(true)
                                        .createTime(LocalDateTime.now())
                                        .updateTime(LocalDateTime.now())
                                        .bizType("QA")
                                        .build();
                                likedRecords.add(record);
                            }
                            log.debug("处理bizId {} 的点赞记录: {} 条", bizId, userIds.size());
                        }
                    } catch (NumberFormatException e) {
                        log.warn("无效的bizId格式: {}", bizIdStr);
                    }
                }
            }
            // 2. 处理取消点赞记录
            if (likedRecords.size() < maxBizSize) {
                String unlikePattern = RedisConstants.UNLIKES_BIZ_KEY_PREFIX + "*";
                Set<String> unlikeKeys = redisTemplate.keys(unlikePattern);
                log.debug("找到 {} 个取消点赞记录的Redis键", CollUtils.isEmpty(unlikeKeys) ? 0 : unlikeKeys.size());
                if (CollUtils.isNotEmpty(unlikeKeys)) {
                    for (String key : unlikeKeys) {
                        String bizIdStr = key.substring(RedisConstants.UNLIKES_BIZ_KEY_PREFIX.length());
                        try {
                            Long bizId = Long.valueOf(bizIdStr);
                            List<String> userIds = redisTemplate.opsForSet().pop(key, maxBizSize - likedRecords.size());
                            if (userIds != null && !userIds.isEmpty()) {
                                for (String userIdStr : userIds) {
                                    Long userId = Long.valueOf(userIdStr);
                                    LikedRecordsDTO record = LikedRecordsDTO.builder()
                                            .userId(userId)
                                            .bizId(bizId)
                                            .liked(false)
                                            .createTime(LocalDateTime.now())
                                            .updateTime(LocalDateTime.now())
                                            .bizType("QA")
                                            .build();
                                    likedRecords.add(record);
                                }
                                log.debug("处理bizId {} 的取消点赞记录: {} 条", bizId, userIds.size());
                            }
                        } catch (NumberFormatException e) {
                            log.warn("无效的bizId格式: {}", bizIdStr);
                        }
                        if (likedRecords.size() >= maxBizSize) {
                            break;
                        }
                    }
                }
            }
            if (CollUtils.isEmpty(likedRecords)) {
                log.debug("没有找到需要同步的点赞记录");
                // 暂停定时任务十分钟
                likedCheckTask.pauseTask();
                return;
            }
            // 3. 发送MQ消息
            try {
                mqHelper.send(
                        LIKE_RECORD_EXCHANGE,
                        StringUtils.format(LIKED_RECORDS_KEY_TEMPLATE, bizType),
                        likedRecords);
                log.info("成功发送记录MQ消息，数据量: {} 条", likedRecords.size());
            } catch (Exception e) {
                log.error("发送记录MQ消息失败，数据量: {} 条", likedRecords.size(), e);
                restoreRecordsToRedis(likedRecords);
            }
        } catch (Exception e) {
            log.error("处理点赞记录同步失败 - 业务类型: {}", bizType, e);
        }
    }

    /**
     * 更高效的批量删除方法（使用IN语句）
     */
    @Override
    public int batchDeleteByUserIdAndBizId(List<LikedRecord> records) {
        if (CollUtils.isEmpty(records)) {
            return 0;
        }
        try {
            // 构建userId和bizId的配对条件
            List<Map<String, Object>> conditions = new ArrayList<>();
            for (LikedRecord record : records) {
                Map<String, Object> condition = new HashMap<>();
                condition.put("userId", record.getUserId());
                condition.put("bizId", record.getBizId());
                conditions.add(condition);
            }
            // 使用自定义Mapper方法进行批量删除
            return baseMapper.batchDeleteByConditions(conditions);
        } catch (Exception e) {
            log.error("批量删除点赞记录失败", e);
            return 0;
        }
    }

    /**
     * 发送失败时恢复数据到Redis
     */
    private void restoreRecordsToRedis(List<LikedRecordsDTO> records) {
        try {
            int restoredCount = 0;
            for (LikedRecordsDTO record : records) {
                String key;
                if (Boolean.TRUE.equals(record.getLiked())) {
                    key = RedisConstants.LIKES_BIZ_KEY_PREFIX + record.getBizId();
                } else {
                    key = RedisConstants.UNLIKES_BIZ_KEY_PREFIX + record.getBizId();
                }
                redisTemplate.opsForSet().add(key, record.getUserId().toString());
                restoredCount++;
            }
            log.info("成功恢复 {} 条记录到Redis", restoredCount);
        } catch (Exception e) {
            log.error("恢复记录到Redis失败", e);
        }
    }

    /**
     * 冷启动问题：Redis启动时空的，但数据库中已有点赞数据
     * 数据不一致：用户取消点赞时，Redis中没有对应记录，但数据库中有
     * 操作丢失：取消点赞操作无法正确同步到数据库
     * <p>
     * 解决方案：不管是点赞还是取消点赞，都写在redis中，之后统一写入数据库liked_record：点赞就是添加新的点赞记录，取消点赞就是删除对应的点赞记录
     */
    private boolean unlike(LikeRecordFormDTO recordDTO) {
        // 1.获取用户id
        Long userId = UserContext.getUser();
        // 2.获取Key
        String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + recordDTO.getBizId();
        String unkey = RedisConstants.UNLIKES_BIZ_KEY_PREFIX + recordDTO.getBizId();
        // 3.执行SREM命令
        Long result = redisTemplate.opsForSet().remove(key, userId.toString());
        // 检查集合是否为空，如果为空可以删除key节省空间
        Long size = redisTemplate.opsForSet().size(key);
        if (size != null && size == 0) {
            redisTemplate.delete(key);
        }
        if (result == null || result <= 0) {
            // 表面Redis中没有这条点赞记录，移除失败
            // 添加反向记录，数据库收到该记录会删除对应的点赞记录
            Long unResult = redisTemplate.opsForSet().add(unkey, userId.toString());
            if (unResult != null && unResult > 0) {
                // 如果反向记录添加成功，设置有效期
                redisTemplate.expire(unkey, RedisConstants.REDIS_EXPIRE_TIME_MINUTES + new Random().nextInt(12), TimeUnit.MINUTES);
            }
        }
        return true;
    }

    private boolean like(LikeRecordFormDTO recordDTO) {
        // 1.获取用户id
        Long userId = UserContext.getUser();
        // KEY（bizId） VALUE(userId)
        // 2.获取Key
        String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + recordDTO.getBizId();
        // 3.执行SADD命令
        Long result = redisTemplate.opsForSet().add(key, userId.toString());
        if (result != null && result > 0) {
            // 设置过期时间（只有新增时才设置，避免覆盖已有的过期时间）
            redisTemplate.expire(key, RedisConstants.REDIS_EXPIRE_TIME_MINUTES + new Random().nextInt(12), TimeUnit.MINUTES);
            return true;
        }
        return false;
    }
}
